// 实现语音识别子服务
#include <brpc/server.h>
#include <unordered_map>
#include <unordered_set>

#include <butil/logging.h>

#include "es_factory.hpp"
#include "es_message.hpp"
#include "mysql_message.hpp" // mysql数据管理客户端封装
#include "etcd.hpp"          // 服务注册模块封装
#include "logger.hpp"        // 日志模块封装
#include "utils.hpp"         // 基础工具接口
#include "channel.hpp"       // 信道管理模块封装
#include "rabbitmq.hpp"

#include "message.pb.h" // protobuf框架代码
#include "base.pb.h"    // protobuf框架代码
#include "file.pb.h"    // protobuf框架代码
#include "user.pb.h"    // protobuf框架代码

class MessageServiceImpl : public lch_im::MsgStorageService
{
public:
    MessageServiceImpl(const shared_ptr<elasticlient::Client> &es_client,
                       const std::shared_ptr<odb::core::database> &mysql_client,
                       const ServiceManager::ptr &channel_manager,
                       const std::string &user_service_name,
                       const std::string &file_service_name) : _es_message(std::make_shared<ESMessage>(es_client)),
                                                               _mysql_message(std::make_shared<MessageTable>(mysql_client)),
                                                               _file_service_name(file_service_name), _user_service_name(user_service_name), _manager_channels(channel_manager)
    {
        _es_message->createIndex();
    }

    void GetHistoryMsg(::google::protobuf::RpcController *controller,
                       const ::lch_im::GetHistoryMsgReq *request,
                       ::lch_im::GetHistoryMsgRsp *response,
                       ::google::protobuf::Closure *done)
    {
        brpc::ClosureGuard rpc_guard(done);
        auto err_response = [this, response](const std::string &rid, const std::string &errmsg) -> void
        {
            response->set_request_id(rid);
            response->set_success(false);
            response->set_errmsg(errmsg);
        };

        // 提取会话ID，开始和结束时间
        std::string rid = request->request_id();
        std::string chat_session_id = request->chat_session_id();
        boost::posix_time::ptime stime = boost::posix_time::from_time_t(request->start_time());
        boost::posix_time::ptime etime = boost::posix_time::from_time_t(request->over_time());
        // 从数据库中查询
        auto msg_lists = _mysql_message->range(chat_session_id, stime, etime);
        if (msg_lists.empty())
        {
            response->set_request_id(rid);
            response->set_success(true);
            return;
        }

        // 统计所有文件的文件ID，并从文件子服务中批量文件下载
        std::unordered_set<std::string> file_id_lists;
        for (auto &msg : msg_lists)
        {
            if (msg.file_id().empty())
                continue;
            LOG_INFO("需要下载的文件ID: {}", msg.file_id());
            file_id_lists.insert(msg.file_id());
        }

        std::unordered_map<std::string, std::string> file_data_lists;
        // 调用_GetFile后 file_data_lists已经有了文件ID和对应的数据
        bool ret = _GetFile(rid, file_id_lists, file_data_lists);
        if (ret == false)
        {
            LOG_ERROR("{} 批量数据下载失败", rid);
            return err_response(rid, "批量数据下载失败");
        }

        // 统计所有消息发送者的用户ID，从用户子服务进行批量获取
        std::unordered_set<std::string> user_id_lists;
        for (auto &msg : msg_lists)
        {
            if (msg.user_id().empty())
                continue;
            LOG_DEBUG("{}",msg.user_id());
            user_id_lists.insert(msg.user_id());
        }
        std::unordered_map<std::string, lch_im::UserInfo> user_lists;
        ret = _GetUser(rid, user_id_lists, user_lists);
        if (ret == false)
        {
            LOG_ERROR("{} 批量用户数据获取失败！", rid);
            return err_response(rid, "批量用户数据获取失败!");
        }

        response->set_request_id(rid);
        response->set_success(true);
        for (const auto &msg : msg_lists)
        {
            auto message_info = response->add_msg_list();
            message_info->set_message_id(msg.message_id());
            message_info->set_chat_session_id(msg.session_id());
            message_info->set_timestamp(boost::posix_time::to_time_t(msg.create_time()));
            message_info->mutable_sender()->CopyFrom(user_lists[msg.user_id()]);
            switch (msg.message_type())
            {
            case lch_im::MessageType::STRING:
                message_info->mutable_message()->set_message_type(lch_im::MessageType::STRING);
                message_info->mutable_message()->mutable_string_message()->set_content(msg.content());
                break;
            case lch_im::MessageType::IMAGE:
                message_info->mutable_message()->set_message_type(lch_im::MessageType::IMAGE);
                message_info->mutable_message()->mutable_image_message()->set_file_id(msg.file_id());
                message_info->mutable_message()->mutable_image_message()->set_image_content(msg.content());
                break;
            case lch_im::MessageType::FILE:
                message_info->mutable_message()->set_message_type(lch_im::MessageType::FILE);
                message_info->mutable_message()->mutable_file_message()->set_file_id(msg.file_id());
                message_info->mutable_message()->mutable_file_message()->set_file_size(msg.file_size());
                message_info->mutable_message()->mutable_file_message()->set_file_name(msg.file_name());
                message_info->mutable_message()->mutable_file_message()->set_file_contents(msg.content());
                break;
            case lch_im::MessageType::SPEECH:
                message_info->mutable_message()->set_message_type(lch_im::MessageType::SPEECH);
                message_info->mutable_message()->mutable_speech_message()->set_file_id(msg.file_id());
                message_info->mutable_message()->mutable_speech_message()->set_speech_contents(msg.content());
                break;
            default:
                LOG_ERROR("消息类型错误");
                break;
            }
        }
        return;
    }

    void GetRecentMsg(::google::protobuf::RpcController *controller,
                      const ::lch_im::GetRecentMsgReq *request,
                      ::lch_im::GetRecentMsgRsp *response,
                      ::google::protobuf::Closure *done)
    {
        brpc::ClosureGuard rpc_guard(done);
        auto err_response = [this, response](const std::string &rid, const std::string &errmsg) -> void
        {
            response->set_request_id(rid);
            response->set_success(false);
            response->set_errmsg(errmsg);
        };

        std::string rid = request->request_id();
        std::string chat_session_id = request->chat_session_id();
        int msg_count = request->msg_count();
        // 从数据库中获取数据
        auto msg_lists = _mysql_message->recent(chat_session_id, msg_count);
        if (msg_lists.empty())
        {
            response->set_request_id(rid);
            response->set_success(true);
            return;
        }

        // 统计所有文件的文件ID，并从文件子服务中批量文件下载
        std::unordered_set<std::string> file_id_lists;
        for (auto &msg : msg_lists)
        {
            if (msg.file_id().empty())
                continue;
            LOG_INFO("需要下载的文件ID: {}", msg.file_id());
            file_id_lists.insert(msg.file_id());
        }

        std::unordered_map<std::string, std::string> file_data_lists;
        // 调用_GetFile后 file_data_lists已经有了文件ID和对应的数据
        bool ret = _GetFile(rid, file_id_lists, file_data_lists);
        if (ret == false)
        {
            LOG_ERROR("{} 批量数据下载失败", rid);
            return err_response(rid, "批量数据下载失败");
        }

        // 统计所有消息发送者的用户ID，从用户子服务进行批量获取
        std::unordered_set<std::string> user_id_lists;
        for (auto &msg : msg_lists)
        {
            if (msg.user_id().empty())
                continue;
            user_id_lists.insert(msg.user_id());
        }
        std::unordered_map<std::string, lch_im::UserInfo> user_lists;
        ret = _GetUser(rid, user_id_lists, user_lists);
        if (ret == false)
        {
            LOG_ERROR("{} 批量用户数据获取失败！", rid);
            return err_response(rid, "批量用户数据获取失败!");
        }

        response->set_request_id(rid);
        response->set_success(true);
        for (const auto &msg : msg_lists)
        {
            auto message_info = response->add_msg_list();
            message_info->set_message_id(msg.message_id());
            message_info->set_chat_session_id(msg.session_id());
            message_info->set_timestamp(boost::posix_time::to_time_t(msg.create_time()));
            message_info->mutable_sender()->CopyFrom(user_lists[msg.user_id()]);
            switch (msg.message_type())
            {
            case lch_im::MessageType::STRING:
                message_info->mutable_message()->set_message_type(lch_im::MessageType::STRING);
                message_info->mutable_message()->mutable_string_message()->set_content(msg.content());
                break;
            case lch_im::MessageType::IMAGE:
                message_info->mutable_message()->set_message_type(lch_im::MessageType::IMAGE);
                message_info->mutable_message()->mutable_image_message()->set_file_id(msg.file_id());
                message_info->mutable_message()->mutable_image_message()->set_image_content(msg.content());
                break;
            case lch_im::MessageType::FILE:
                message_info->mutable_message()->set_message_type(lch_im::MessageType::FILE);
                message_info->mutable_message()->mutable_file_message()->set_file_id(msg.file_id());
                message_info->mutable_message()->mutable_file_message()->set_file_size(msg.file_size());
                message_info->mutable_message()->mutable_file_message()->set_file_name(msg.file_name());
                message_info->mutable_message()->mutable_file_message()->set_file_contents(msg.content());
                break;
            case lch_im::MessageType::SPEECH:
                message_info->mutable_message()->set_message_type(lch_im::MessageType::SPEECH);
                message_info->mutable_message()->mutable_speech_message()->set_file_id(msg.file_id());
                message_info->mutable_message()->mutable_speech_message()->set_speech_contents(msg.content());
                break;
            default:
                LOG_ERROR("消息类型错误");
                break;
            }
        }
    }

    void MsgSearch(::google::protobuf::RpcController *controller,
                   const ::lch_im::MsgSearchReq *request,
                   ::lch_im::MsgSearchRsp *response,
                   ::google::protobuf::Closure *done)
    {
        brpc::ClosureGuard rpc_guard(done);
        auto err_response = [this, response](const std::string &rid, const std::string &errmsg) -> void
        {
            response->set_request_id(rid);
            response->set_success(false);
            response->set_errmsg(errmsg);
        };

        std::string rid = request->request_id();
        std::string chat_session_id = request->chat_session_id();
        std::string search_key = request->search_key();

        auto msg_lists = _es_message->search(search_key, chat_session_id);
        if (msg_lists.empty())
        {
            response->set_request_id(rid);
            response->set_success(true);
            return;
        }

        // 组织所有消息的用户ID，从用户子服务获取用户信息
        std::unordered_set<std::string> user_id_lists;
        for (const auto &msg : msg_lists)
        {
            user_id_lists.insert(msg.user_id());
        }
        std::unordered_map<std::string, lch_im::UserInfo> user_lists;
        bool ret = _GetUser(rid, user_id_lists, user_lists);
        if (ret == false)
        {
            LOG_ERROR("{} 批量用户数据获取失败！", rid);
            return err_response(rid, "批量用户数据获取失败!");
        }
        // 组织响应
        response->set_request_id(rid);
        response->set_success(true);
        for (const auto &msg : msg_lists)
        {
            auto message_info = response->add_msg_list();
            message_info->set_message_id(msg.message_id());
            message_info->set_chat_session_id(msg.session_id());
            message_info->set_timestamp(boost::posix_time::to_time_t(msg.create_time()));
            message_info->mutable_sender()->CopyFrom(user_lists[msg.user_id()]);
            // 注意这里只有文本消息，因为搜索的时候只搜索文本消息
            message_info->mutable_message()->set_message_type(lch_im::MessageType::STRING);
            message_info->mutable_message()->mutable_string_message()->set_content(msg.content());
        }
        return;
    }

    void onMessage(const char *body, size_t sz)
    {
        LOG_ERROR("mq收到消息，开始进行存储");
        // 取出消息，进行反序列化
        lch_im::MessageInfo message;
        bool ret = message.ParseFromArray(body, sz);
        if (ret == false)
        {
            if ("消息反序列化失败");
            return;
        }

        // 针对不同的消息进行不同的处理
        std::string file_id;
        int64_t file_size;
        std::string file_name;
        std::string content;
        switch (message.message().message_type())
        {
        // 如果是文本消息，向es进行存储
        case lch_im::MessageType::STRING:
            content = message.message().string_message().content();
            ret = _es_message->appendData(
                message.sender().user_id(),
                message.message_id(),
                message.timestamp(),
                message.chat_session_id(),
                content);

            if (ret == false)
            {
                LOG_ERROR("文本消息向es存储失败");
                return;
            }

            break;
        // 如果是语音 文件 图片消息，就存储到文件子服务中，并获取文件ID
        case lch_im::MessageType::IMAGE:
        {
            const auto &msg = message.message().image_message();
            ret = _PutFile("", msg.image_content(), msg.image_content().size(), file_id);
            if (ret == false)
            {
                LOG_ERROR("上传图片到文件子服务失败");
                return;
            }
        }
        break;
        case lch_im::MessageType::FILE:
        {
            const auto &msg = message.message().file_message();
            file_size = msg.file_size();
            file_name = msg.file_name();
            ret = _PutFile(file_name, msg.file_contents(), msg.file_size(), file_id);
            if (ret == false)
            {
                LOG_ERROR("上传文件到文件子服务失败");
                return;
            }
        }
        break;
        case lch_im::MessageType::SPEECH:
        {
            const auto &msg = message.message().speech_message();
            ret = _PutFile("", msg.speech_contents(), msg.speech_contents().size(), file_id);
            if (ret == false)
            {
                LOG_ERROR("上传文件到文件子服务失败");
                return;
            }
        }
        break;
        default:
            LOG_ERROR("消息类型错误！");
            return;
        }
        // 获取消息信息，存储到mysql中
        Message msg(message.message_id(),
                    message.chat_session_id(),
                    message.sender().user_id(),
                    message.message().message_type(),
                    boost::posix_time::from_time_t(message.timestamp()));
        msg.set_content(content);
        msg.set_file_id(file_id);
        msg.set_file_name(file_name);
        msg.set_file_size(file_size);
        ret = _mysql_message->insert(msg);
        if (ret == false)
        {
            LOG_ERROR("向数据库插入新消息失败！");
            return;
        }
    }

    // 文件的上传
    bool _PutFile(const std::string &filename,
                  const std::string &fbody,
                  const int64_t &fsize,
                  std::string &file_id)
    {
        auto channel = _manager_channels->choose(_file_service_name);
        if (!channel)
        {
            LOG_ERROR("{} 没有可供访问的文件子服务节点！", _file_service_name);
            return false;
        }

        lch_im::FileService_Stub stub(channel.get());
        lch_im::PutSingleFileReq req;
        lch_im::PutSingleFileRsp rsp;
        req.mutable_file_data()->set_file_name(filename);
        req.mutable_file_data()->set_file_size(fsize);
        req.mutable_file_data()->set_file_content(fbody);
        brpc::Controller cntl;
        stub.PutSingleFile(&cntl, &req, &rsp, nullptr);
        if (cntl.Failed() == true || rsp.success() == false)
        {
            LOG_ERROR("文件子服务调用失败: {}", cntl.ErrorText());
            return false;
        }

        file_id = rsp.file_info().file_id();
        return true;
    }

    // 文件的获取
    bool _GetFile(const std::string &rid,
                  const std::unordered_set<std::string> &file_id_lists,
                  std::unordered_map<std::string, std::string> &file_data_lists)
    {
        auto channel = _manager_channels->choose(_file_service_name);
        if (!channel)
        {
            LOG_ERROR("{} 没有可供访问的文件子服务节点！", _file_service_name);
            return false;
        }

        lch_im::FileService_Stub stub(channel.get());
        lch_im::GetMultiFileReq req;
        lch_im::GetMultiFileRsp rsp;
        req.set_request_id(rid);
        for (const auto &id : file_id_lists) {
            req.add_file_id_list(id);
        }

        brpc::Controller cntl;
        stub.GetMultiFile(&cntl, &req, &rsp, nullptr);
        if (cntl.Failed() == true || rsp.success() == false)
        {
            LOG_ERROR("{} 文件子服务调用失败: {}", rid, cntl.ErrorText());
            return false;
        }

        const auto &fmap = rsp.file_data();
        for (auto it = fmap.begin(); it != fmap.end(); it++)
        {
            file_data_lists.insert(std::make_pair(it->first, it->second.file_content()));
        }

        return true;
    }

    bool _GetUser(const std::string &rid,
                  const std::unordered_set<std::string> &user_id_lists,
                  std::unordered_map<std::string, lch_im::UserInfo> &user_lists)
    {
        auto channel = _manager_channels->choose(_user_service_name);
        if (!channel)
        {
            LOG_ERROR("{} 没有可供访问的文件子服务节点！", _file_service_name);
            return false;
        }

        lch_im::UserService_Stub stub(channel.get());
        lch_im::GetMultiUserInfoReq req;
        lch_im::GetMultiUserInfoRsp rsp;
        req.set_request_id(rid);
        for (auto &id : user_id_lists)
        {
            req.add_users_id(id);
        }
      
        brpc::Controller cntl;
        stub.GetMultiUserInfo(&cntl, &req, &rsp, nullptr);
        if (cntl.Failed() == true || rsp.success() == false)
        {
            LOG_ERROR(" 文件子服务调用失败: {}", cntl.ErrorText());
            return false;
        }

        const auto &umap = rsp.users_info();
        for (auto it = umap.begin(); it != umap.end(); ++it)
        {
            user_lists.insert(std::make_pair(it->first, it->second));
        }
        return true;
    }

    ~MessageServiceImpl() {}

private:
    ESMessage::ptr _es_message;
    MessageTable::ptr _mysql_message;
    // 这边是rpc调用客户端相关对象
    std::string _user_service_name;
    std::string _file_service_name;
    ServiceManager::ptr _manager_channels;
};

class MessageServer
{
public:
    using ptr = std::shared_ptr<MessageServer>;
    MessageServer(const MQClient::ptr &mq_client,
                  const Discovery::ptr &service_discoverer,
                  const Registery::ptr &registry_client,
                  const std::shared_ptr<elasticlient::Client> &es_client,
                  const std::shared_ptr<odb::core::database> mysql_client,
                  const std::shared_ptr<brpc::Server> &rpc_server)
        : _service_discoverer(service_discoverer),
          _registry_client(registry_client),
          _mq_client(mq_client),
          _es_client(es_client),
          _mysql_client(mysql_client),
          _rpc_server(rpc_server)
    {
    }

    ~MessageServer() {}

    void start()
    {
        _rpc_server->RunUntilAskedToQuit();
    }

private:
    Discovery::ptr _service_discoverer;
    Registery::ptr _registry_client;
    MQClient::ptr _mq_client;
    std::shared_ptr<elasticlient::Client> _es_client;
    std::shared_ptr<odb::core::database> _mysql_client;
    std::shared_ptr<brpc::Server> _rpc_server;
};

class MessageServerBuild
{
public:
    // 创建es对象
    void create_es_object(const std::vector<std::string> host_list)
    {
        _es_client = ESClientFactory::create(host_list);
    }
    // 创建mysql对象
    void create_mysql_object(
        const std::string &user,
        const std::string &pswd,
        const std::string &db,
        const std::string &host,
        const std::string &cset,
        int port,
        int connect_pool_count)
    {
        _mysql_client = ODBFactory::create(user, pswd, db, host, cset, port, connect_pool_count);
    }

    // 用于构造服务发现客户端&信道管理对象
    void create_discovery_object(const std::string &reg_host,
                               const std::string &base_service_name,
                               const std::string &file_service_name,
                               const std::string &user_service_name)
    {
        _user_service_name = user_service_name;
        _file_service_name = file_service_name;
        _manager_channels = std::make_shared<ServiceManager>();
        _manager_channels->declared(file_service_name);
        _manager_channels->declared(user_service_name);
        LOG_DEBUG("设置文件子服务为需添加管理的子服务：{}", file_service_name);
        auto put_cb = std::bind(&ServiceManager::onServiceOnline, _manager_channels.get(), std::placeholders::_1, std::placeholders::_2);
        auto del_cb = std::bind(&ServiceManager::onServiceOffline, _manager_channels.get(), std::placeholders::_1, std::placeholders::_2);
        _service_discoverer = std::make_shared<Discovery>(reg_host, base_service_name, put_cb, del_cb);
    }

    void create_registry_object(const std::string &reg_host, const std::string &service_name, const std::string &access_host)
    {
        _registry_client = std::make_shared<Registery>(reg_host);
        _registry_client->registry(service_name, access_host);
    }

    // 构造rabbitmq客户端对象
    void create_mq_object(const std::string &user,
                          const std::string &passwd,
                          const std::string &host,
                          const std::string &exchange_name,
                          const std::string &queue_name,
                          const std::string &binding_key)
    {
        _exchange_name = exchange_name;
        _mq_client = std::make_shared<MQClient>(user, passwd, host);
        _mq_client->declareComponents(exchange_name, queue_name, binding_key);
    }

    void create_rpc_server(uint16_t port, int32_t timeout, uint8_t num_threads)
    {
        if (!_es_client)
        {
            LOG_ERROR("还未初始化ES搜索引擎模块！");
            abort();
        }

        if (!_mysql_client)
        {
            LOG_ERROR("还未初始化mysql客户端");
            abort();
        }

        if (!_manager_channels)
        {
            LOG_ERROR("还未初始化信道管理模块！");
            abort();
        }
        _rpc_server = std::make_shared<brpc::Server>();
        MessageServiceImpl *msg_service = new MessageServiceImpl(
            _es_client, _mysql_client, _manager_channels, _user_service_name, _file_service_name);

        int ret = _rpc_server->AddService(msg_service,
                                          brpc::ServiceOwnership::SERVER_OWNS_SERVICE);
        if (ret == -1)
        {
            LOG_ERROR("添加Rpc服务失败！");
            abort();
        }

        // 启动服务器
        brpc::ServerOptions options;
        // 连接空闲超时时间，-1表示
        options.idle_timeout_sec = -1;
        // io线程数量
        options.num_threads = 1;
        ret = _rpc_server->Start(port, &options);
        if (ret == -1)
        {
            LOG_ERROR("服务启动失败");
            abort();
        }

        auto callback = std::bind(&MessageServiceImpl::onMessage,msg_service,std::placeholders::_1,std::placeholders::_2);
        _mq_client->consume(_queue_name,callback);
    }

    MessageServer::ptr build()
    {
        if (!_service_discoverer)
        {
            LOG_ERROR("没有初始化服务发现服务");
            abort();
        }

        if (!_registry_client)
        {
            LOG_ERROR("没有初始化服务注册服务");
            abort();
        }

        if (!_rpc_server)
        {
            LOG_ERROR("没有初始化rpc服务");
            abort();
        }

        MessageServer::ptr server = std::make_shared<MessageServer>(_mq_client,
                                                                    _service_discoverer,
                                                                    _registry_client,
                                                                    _es_client,
                                                                    _mysql_client,
                                                                    _rpc_server);
        return server;
    }

private:
    // 服务注册
    Registery::ptr _registry_client;

    std::shared_ptr<elasticlient::Client> _es_client;
    std::shared_ptr<odb::core::database> _mysql_client;

    std::string _user_service_name;
    std::string _file_service_name;
    ServiceManager::ptr _manager_channels;
    Discovery::ptr _service_discoverer;

    std::string _exchange_name;
    std::string _queue_name;
    MQClient::ptr _mq_client;
    // rpc
    std::shared_ptr<brpc::Server> _rpc_server;
};
